package com.fwmagic.flink.outputtag;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * （连续）分流操作
 * 定义好 OutputTag 后，可以使用下面几种函数来处理数据：
 * <p>
 * ProcessFunction
 * KeyedProcessFunction
 * CoProcessFunction
 * ProcessWindowFunction
 * ProcessAllWindowFunction
 */
public class OutputTagExample {

    private static final String[] WORDS = {"aaa", "bbb", "ccc", "ddd", "abc", "bcd", "cde", "dfg","eee"};

    private static final OutputTag<String> ATAG = new OutputTag<String>("A") {
    };
    private static final OutputTag<String> BTAG = new OutputTag<String>("B") {
    };
    private static final OutputTag<String> CTAG = new OutputTag<String>("C") {
    };
    private static final OutputTag<String> DTAG = new OutputTag<String>("D") {
    };

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataSource = env.fromElements(WORDS);

        SingleOutputStreamOperator<String> sideOutPutTagData
                = dataSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String word, Context ctx, Collector<String> out) throws Exception {
                if (word.startsWith("a")) {
                    ctx.output(ATAG, word);
                } else if (word.startsWith("b")) {
                    ctx.output(BTAG, word);
                } else if (word.startsWith("c")) {
                    ctx.output(CTAG, word);
                } else if (word.startsWith("d")) {
                    ctx.output(DTAG, word);
                }else{
                    out.collect(word);
                }
            }
        });

        DataStream<String> tagA = sideOutPutTagData.getSideOutput(ATAG);
        DataStream<String> tagB = sideOutPutTagData.getSideOutput(BTAG);
        DataStream<String> tagC = sideOutPutTagData.getSideOutput(CTAG);
        DataStream<String> tagD = sideOutPutTagData.getSideOutput(DTAG);

        //tagA.print();
        //tagB.print();
        //tagC.print();
        //tagD.print();

        //eee
        sideOutPutTagData.print();

        env.execute("fwmagic sideOutPutTag example");

    }
}
